package com.echo.im.server.netty.processor.impl;


import com.echo.im.common.enums.IMCmdType;
import com.echo.im.common.model.IMRecvInfo;
import com.echo.im.common.model.IMSendInfo;
import com.echo.im.common.model.IMUserInfo;
import com.echo.im.common.model.mq.OfflineMsgMqInfo;
import com.echo.im.common.vo.message.SystemMessageVO;
import com.echo.im.mq.event.model.OfflineMessageEvent;
import com.echo.im.mq.event.producer.EventPublisher;
import com.echo.im.server.ack.MessageAckManager;
import com.echo.im.server.netty.context.UserChannelMap;
import com.echo.im.server.netty.processor.AbstractMessageProcessor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

/**
 *
 * @author echo
 * @date 2024-08-14
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class SystemMessageProcessor extends AbstractMessageProcessor<IMRecvInfo<SystemMessageVO>> {

    private final EventPublisher eventPublisher;
    private final OfflineMessageEvent offlineMessageEvent;
    private final MessageAckManager messageAckManager;

    @Override
    public void process(ChannelHandlerContext ctx, IMRecvInfo<SystemMessageVO> data) {

    }

    @Override
    public void process(IMRecvInfo<SystemMessageVO> recvInfo) {
        List<IMUserInfo> receivers = recvInfo.getReceivers();
        for (IMUserInfo receiver : receivers) {
            SystemMessageVO systemMessageVO = recvInfo.getData();
            log.info("接收到系统消息,接收者:{}，内容:{}", receiver.getId(), systemMessageVO);
            try {
                Channel channel = UserChannelMap.getChannel(receiver.getId(), receiver.getTerminal());

                //Channel为null，则证明用户不在线，则消息入离线消息队列
                if (Objects.isNull(channel)){
                    OfflineMsgMqInfo offlineMsgMqInfo = new OfflineMsgMqInfo(receiver.getId(), recvInfo.getData().getId(),OfflineMsgMqInfo.OfflineMsgEnum.SYSTEM_OFFLINE_MESSAGE);
                    eventPublisher.publish(offlineMessageEvent.topic(),offlineMessageEvent.buildEventMessage(offlineMsgMqInfo));
                    return;
                }

                //推送消息
                IMSendInfo<Object> sendInfo = new IMSendInfo<>();
                sendInfo.setCmd(IMCmdType.SYSTEM_MESSAGE.code());
                sendInfo.setData(systemMessageVO);
                channel.writeAndFlush(sendInfo);

                //并存入 ack 等待队列，等待接受者的 ack，如果超时未收到 ack，直接断开连接，让用户拉取离线消息
                //ack队列里面只应该存储用户和消息id，只要用户的所有在线终端，有一个收到了消息，就够了，如果超过3s未收到ack，直接断开连接，并将消息加入离线队列
                messageAckManager.addPendingAck(recvInfo.getData().getId(), receiver.getId(), channel.id().asLongText(), OfflineMsgMqInfo.OfflineMsgEnum.SYSTEM_OFFLINE_MESSAGE);


            } catch (Exception e) {
                //  直接将消息加入离线消息
                OfflineMsgMqInfo offlineMsgMqInfo = new OfflineMsgMqInfo(receiver.getId(), recvInfo.getData().getId(),OfflineMsgMqInfo.OfflineMsgEnum.SYSTEM_OFFLINE_MESSAGE);
                eventPublisher.publish(offlineMessageEvent.topic(),offlineMessageEvent.buildEventMessage(offlineMsgMqInfo));
                log.error("发送异常，接收者:{}，内容:{}", receiver.getId(), systemMessageVO, e);
            }
        }
    }

    @Override
    public IMCmdType getCmdType() {
        return IMCmdType.SYSTEM_MESSAGE;
    }

    @Override
    public IMRecvInfo transForm(Object o) {
        return null;
    }
}
